package com.bbx.flink.demo.coll;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 从集合中获取数据源
 */
@Slf4j
public class ReadCollectDemo {

//    /**
//     *   //方式一：  fromCollection(Collection)  从java 集合中获取数据源，所有元素必须是同一类型
//     * @param args
//     * @throws Exception
//     */
//    public  static void  main (String [] args) throws Exception {
//        //创建执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        List<String> list = new ArrayList<>();
//        list.add("a");
//        list.add("b");
//        list.add("c");
//        list.add("d");
//        list.add("e");
//        list.add("f");
//        list.add("g");
//        list.add("h");
//        list.add("w");
//        //获取数据源
//        //方式一：  fromCollection(Collection)  从java 集合中获取数据源，所有元素必须是同一类型
//        DataStreamSource<String> streamSource = env.fromCollection(list);
//        //数据输出
//        streamSource.print();
//        //开始执行
//        env.execute();
//    }



//    /**
//     * Creates a new data stream that contains elements in the iterator
//     *   //方式二：  fromParallelCollection(SplittableIterator<OUT> iterator, Class<OUT> type)
//     *
//     * @param args
//     * @throws Exception
//     */
//    public  static void  main (String [] args) throws Exception {
//        //创建执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        //获取数据源
//        Iterator<Student> iterator = set.iterator();
//        NumberSequenceIterator numberSequenceIterator = new NumberSequenceIterator(10,50);
//        DataStreamSource<Long> streamSource = env.fromParallelCollection(numberSequenceIterator, Long.class);
//        //数据输出
//        streamSource.print();
//        //开始执行
//        env.execute();
//    }


//    /**
//     * 从迭代器创建数据流源，
//     * 注意：不能通过集合的 iterator() 方法来获取集合，因为获取的迭代器没有实现序列化接口，不能进行序列化，
//     * 此处用到的迭代器只能是 实现Iterator和 Serializable 接口的类
//     * 方式三： fromCollection(Iterator<OUT> data, Class<OUT> type)
//     *
//     * @param args
//     * @throws Exception
//     */
//    public  static  void main(String [] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        NumberSequenceIterator numberSequenceIterator = new NumberSequenceIterator(10,50);
//        DataStreamSource<Long> dataStreamSource = env.fromCollection(numberSequenceIterator, Long.class);
//        dataStreamSource.print();
//        env.execute();
//    }


//    /**
//     * 根据给定的元素队列创建数据源
//     * 方式四：fromElements(T ...)
//     *
//     * @param args
//     * @throws Exception
//     */
//    public  static  void main(String [] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        DataStreamSource<String> dataStreamSource = env.fromElements("a", "b", "c");
//        dataStreamSource.print();
//        env.execute();
//    }


    /**
     *
     * 方式五：从指定的数值队列创建数据流，创建的是并行的数据源
     *
     * @param args
     * @throws Exception
     */
    public  static  void main(String [] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Long> dataStreamSource = env.fromSequence(1L,10L);
        dataStreamSource.print();
        env.execute();
    }



}





























